package com.zzp.baidu.export;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zzp.baidu.domain.News;
import com.zzp.baidu.mapper.NewsMapper;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.IndexQuery;

/**
 * 新闻批量插入
 * <p>
 * //TODO
 * NewsExport.java
 * </p>
 *
 * @author 佐斯特勒
 * @version v0.0.1
 * @date 2021/2/14 21:45
 * @see com.zzp.baidu.export
 **/
public class NewsExport {

    private final Logger log = LoggerFactory.getLogger(NewsExport.class);

    private final ElasticsearchRestTemplate elasticsearchRestTemplate;

    private final NewsMapper newsMapper;

    public NewsExport(ElasticsearchRestTemplate elasticsearchRestTemplate, NewsMapper newsMapper) {
        this.elasticsearchRestTemplate = elasticsearchRestTemplate;
        this.newsMapper = newsMapper;
    }

    /**
     * 步进
     */
    private int step = 100;

    /**
     * 设置五个线程池
     */
    private ExecutorService executorService = new ThreadPoolExecutor(5, 10, 60,
            TimeUnit.SECONDS, new LinkedBlockingQueue<>(32), new ThreadPoolExecutor.CallerRunsPolicy());

    /**
     * 初始化插入
     */
    public void initInsert() {
        var min = newsMapper.getMinPrimaryKey();
        var max = newsMapper.getMaxPrimaryKey();
        log.info("索引最小值: " + min + "; 索引最大值: " + max + "; 步进值: " + step);
        // 提供者集合，用来提供news集合
        var suppliers = new ArrayList<Supplier<List<News>>>();
        // 待处理的新闻对象
        for (var intId = min; intId <= max; intId += step) {
            suppliers.add(new NewsSupplier(intId, intId + step - 1, newsMapper));
        }
        log.debug(suppliers.toString());
        // 异步批量插入
        suppliers.forEach(supplier -> CompletableFuture.supplyAsync(supplier, executorService)
                .thenAcceptAsync(news -> elasticsearchRestTemplate.bulkIndex(getEsInsertData(news), IndexCoordinates.of("news"))));
    }


    /**
     * Nws 的查询索引
     *
     * @param list News集合
     * @return List<IndexQuery>
     */
    private List<IndexQuery> getEsInsertData(List<News> list) {
        List<IndexQuery> indexQueryList = new ArrayList<>();
        list.forEach(news -> {
            var om = new ObjectMapper();
            IndexQuery indexQuery = new IndexQuery();
            try {
                indexQuery.setSource(om.writeValueAsString(news));
            } catch (JsonProcessingException e) {
                log.error("序列化news 失败！", e);
            }
            indexQueryList.add(indexQuery);
            log.info("插入:" + news);
        });
        return indexQueryList;
    }

    /**
     * 新闻数据提供者
     */
    private static class NewsSupplier implements Supplier<List<News>> {

        private final NewsMapper newsMapper;

        private final int min;
        private final int max;

        public NewsSupplier(int min, int max, NewsMapper newsMapper) {
            this.min = min;
            this.max = max;
            this.newsMapper = newsMapper;
        }

        @Override
        public List<News> get() {
            return newsMapper.findBetweenId(min, max);
        }
    }

    public NewsExport setStep(int step) {
        this.step = step;
        return this;
    }
}
